package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter=new MemoryDataCenter();
    private DiskDataCenter diskDataCenter=new DiskDataCenter();
    private Router router=new Router();
    private ConsumerManager consumerManager=new ConsumerManager(this);

    private final Object exchangeLocker=new Object();
    private final Object queueLocker=new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName=name;
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        }catch (IOException| MqException|ClassNotFoundException e){
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败");
        }
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,boolean autoDelete, Map<String,Object> arguments){
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange existsExchange=memoryDataCenter.getExchange(exchangeName);
                if(existsExchange!=null){
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName="+exchangeName);
                    return true;
                }
                Exchange exchange=new Exchange();
                exchange.setName(exchangeName);
                exchange.setDurable(durable);
                exchange.setType(exchangeType);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！exchangeName="+exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName){
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                Exchange toDelete=memoryDataCenter.getExchange(exchangeName);
                if(toDelete==null){
                    throw new MqException("[VirtualHost] 不存在无法删除！");
                }
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName="+exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除失败！exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,Map<String,Object> arguments){
        queueName=virtualHostName+queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName="+queueName);
                    return true;
                }
                MSGQueue queue=new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if(queue.isDurable()){
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败！queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDelete(String queueName){
        queueName=virtualHostName+queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue toDelete = memoryDataCenter.getQueue(queueName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 队列不存在无法删除！");
                }
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列删除失败！queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueBind(String queueName,String exchangeName,String bindingKey) throws MqException {
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding existsBinding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding!=null){
                        throw new MqException("[VirtualHost] 已经存在！queueName="+queueName+",exchangeName="+exchangeName);
                    }
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法！bindingKey="+bindingKey);
                    }
                    Binding binding=new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);

                    MSGQueue queue=memoryDataCenter.getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[VirtualHost] 队列不存在！queueName="+queueName);
                    }
                    Exchange exchange=memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw new MqException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
                    }
                    if(queue.isDurable()&&exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功！exchangeName="+exchangeName+",queueName="+queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败，queueName="+queueName+",exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueUnbind(String queueName,String exchangeName){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding toDelete=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(toDelete==null){
                        throw new MqException("[VirtualHost] 绑定不存在！exchangeName="+exchangeName+",queueName="+queueName);
                    }
                    diskDataCenter.deleteBinding(toDelete);
                    memoryDataCenter.deleteBinding(toDelete);
                    System.out.println("[VirtualHost] 删除绑定成功");
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败");
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try {
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey非法！routingKey="+routingKey);
            }
            exchangeName = virtualHostName + exchangeName;
            Exchange exchange=memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
            }
            if(exchange.getType()==ExchangeType.DIRECT){
                String queueName=virtualHostName+routingKey;
                MSGQueue queue=memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在！queueName="+queueName);
                }
                Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                sendMessage(queue,message);
            }else{
                ConcurrentHashMap<String,Binding> bindingsMap=memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String,Binding> entry:bindingsMap.entrySet()){
                    Binding binding=entry.getValue();
                    MSGQueue queue=memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue==null){
                        System.out.println("[VirtualHost] basicPublish发送消息时，发现队列不存在！queueName="+binding.getQueueName());
                        continue;
                    }
                    Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    sendMessage(queue,message);
                }
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 消息发送失败！");
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode=message.getDeliverMode();
        if(deliverMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);
        consumerManager.notifyConsume(queue.getName());
    }

    public boolean basicConsume(String consumerTag,String queueName,boolean autoAck,Consumer consumer){
        queueName=virtualHostName+queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume成功，queueName="+queueName);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicConsume失败，queueName="+queueName);
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName=virtualHostName+queueName;
        try {
            MSGQueue queue=memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 要确认的队列不存在，queueName="+queueName);
            }
            Message message=memoryDataCenter.getMessage(messageId);
            if(message==null){
                throw new MqException("[VirtualHost] 要确认的消息不存在，messageId="+messageId);
            }
            if(message.getDeliverMode()==2){
                diskDataCenter.deleteMessage(queue,message);
            }
            memoryDataCenter.removeMessage(messageId);
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck成功，消息被确认成功，queueName="+queueName+",MessageId="+messageId);
            return true;
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("[VirtualHost] basicAck失败，消息被确认失败，queueName="+queueName+",MessageId="+messageId);
            return false;
        }
    }
}
